package cn.iotab.iotkit.transport.mqtt.session;

import cn.iotab.iotkit.transport.api.common.message.msg.up.SessionEvent;
import cn.iotab.iotkit.transport.api.common.message.msg.up.SessionInfo;
import cn.iotab.iotkit.transport.api.driver.TransportDriverService;
import cn.iotab.iotkit.transport.api.driver.context.SessionContext;
import cn.iotab.iotkit.transport.mqtt.MqttTransportContext;
import cn.iotab.iotkit.transport.mqtt.adaptors.JsonMqttAdaptor;
import cn.iotab.iotkit.transport.mqtt.adaptors.MqttTransportAdaptor;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttQoS;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;

/**
 * @author Andrew Shvayka
 */
@Slf4j
public class MqttSessionCtx extends SessionContext {

    private static final MqttTransportAdaptor mqttTransportAdaptor = new JsonMqttAdaptor();

    private final MqttTransportContext transportContext;
    /**
     * 记录设备订阅topic的QOS，向设备发消息时使用
     */
    private final ConcurrentMap<MqttTopicMatcher, Integer> mqttQoSMap;

    private volatile ChannelHandlerContext channelHandlerContext;

    public MqttSessionCtx(String sessionId, MqttTransportContext transportContext) {
        super(sessionId);
        this.transportContext = transportContext;
        this.mqttQoSMap = new ConcurrentHashMap<>();
    }

    public void setChannelHandlerContext(ChannelHandlerContext channelHandlerContext) {
        this.channelHandlerContext = channelHandlerContext;
    }

    public ChannelHandlerContext getChannelHandlerContext() {
        return channelHandlerContext;
    }

    public MqttTransportAdaptor getPayloadAdaptor() {
        return mqttTransportAdaptor;
    }

    public ConcurrentMap<MqttTopicMatcher, Integer> getMqttQoSMap() {
        return mqttQoSMap;
    }

    public MqttQoS getQoSForTopic(String topic) {
        List<Integer> qosList = mqttQoSMap.entrySet()
                .stream()
                .filter(entry -> entry.getKey().matches(topic))
                .map(Map.Entry::getValue)
                .collect(Collectors.toList());
        if (!qosList.isEmpty()) {
            return MqttQoS.valueOf(qosList.get(0));
        } else {
            return MqttQoS.AT_LEAST_ONCE;
        }
    }

    /**
     * 断开连接
     */
    public void processDisconnect() {
        if (channelHandlerContext != null) {
            //连接之后未发送任何消息，则HandlerContext为null
            channelHandlerContext.close();
        }
        SessionInfo sessionInfo = getSessionInfo();
        String productKey = sessionInfo == null ? "unknown" : sessionInfo.getProductKey();
        String deviceId = sessionInfo == null ? "unknown" : sessionInfo.getDeviceId();
        log.info("[{}] Client {}:{} disconnected!", getSessionId(), productKey, deviceId);
        doDisconnect();
    }

    /**
     * 关闭连接
     */
    private void doDisconnect() {
        if (isConnected()) {
            SessionInfo sessionInfo = getSessionInfo();
            if (sessionInfo != null) {
                TransportDriverService transportDriverService = transportContext.getTransportDriverService();
                transportDriverService.onSessionStateChange(getSessionInfo(), SessionEvent.CLOSED, null);
                transportDriverService.deregisterSession(getSessionInfo());
            }
            setDisconnected();
        }
    }


}
